package com.example.sse.service;

import com.alibaba.fastjson.JSON;
import com.example.sse.model.RealtimeData;
import com.example.sse.util.MapUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * SSE连接管理服务
 * 管理所有SSE连接，支持集群部署
 */
@Slf4j
@Service
public class SseConnectionManager {

    @Value("${server.port:8080}")
    private String serverPort;

    @Value("${sse.connection.timeout:300000}")
    private Long connectionTimeout;

    /**
     * 存储所有活跃的SSE连接
     * Key: 连接ID, Value: SseEmitter
     */
    private final Map<String, SseEmitter> connections = new ConcurrentHashMap<>();

    /**
     * 连接计数器
     */
    private final AtomicLong connectionCounter = new AtomicLong(0);

    /**
     * 心跳检测执行器
     */
    private final ScheduledExecutorService heartbeatExecutor = Executors.newScheduledThreadPool(2);

    /**
     * 节点ID（用于集群识别）
     */
    private String nodeId;

    /**
     * 初始化方法，在依赖注入完成后执行
     */
    @PostConstruct
    public void init() {
        log.info("开始初始化SSE连接管理器...");
        log.info("serverPort值: [{}]", serverPort);
        log.info("connectionTimeout值: [{}]", connectionTimeout);
        
        if (serverPort == null) {
            log.error("serverPort为null，使用默认端口");
            serverPort = "8080";
        }
        
        if (connectionTimeout == null) {
            log.error("connectionTimeout为null，使用默认超时");
            connectionTimeout = 300000L;
        }
        
        this.nodeId = "node-" + serverPort + "-" + System.currentTimeMillis();
        startHeartbeat();
        log.info("SSE连接管理器初始化完成，节点ID: {}", nodeId);
    }

    /**
     * 创建新的SSE连接
     * @param clientId 客户端ID
     * @return SseEmitter
     */
    public SseEmitter createConnection(String clientId) {
        String connectionId = generateConnectionId(clientId);
        
        SseEmitter emitter = new SseEmitter(connectionTimeout);
        
        // 设置连接回调
        emitter.onCompletion(() -> {
            connections.remove(connectionId);
            log.info("SSE连接完成: {}, 剩余连接数: {}", connectionId, connections.size());
        });
        
        emitter.onTimeout(() -> {
            connections.remove(connectionId);
            log.warn("SSE连接超时: {}, 剩余连接数: {}", connectionId, connections.size());
        });
        
        emitter.onError((ex) -> {
            connections.remove(connectionId);
            log.error("SSE连接错误: {}, 错误: {}, 剩余连接数: {}", 
                connectionId, ex.getMessage(), connections.size());
        });
        
        // 存储连接
        connections.put(connectionId, emitter);
        
        // 发送欢迎消息
        sendWelcomeMessage(emitter, connectionId);
        
        log.info("新SSE连接建立: {}, 总连接数: {}", connectionId, connections.size());
        
        return emitter;
    }

    /**
     * 向所有连接广播消息
     * @param data 要广播的数据
     */
    public void broadcast(RealtimeData data) {
        if (connections.isEmpty()) {
            log.debug("没有活跃连接，跳过广播");
            return;
        }

        data.setNodeId(nodeId);
        String jsonMessage = JSON.toJSONString(data);
        
        log.info("广播消息到 {} 个连接: {}", connections.size(), data.getType());
        
        connections.entrySet().removeIf(entry -> {
            try {
                entry.getValue().send(SseEmitter.event()
                    .id(data.getId())
                    .name(data.getType())
                    .data(jsonMessage));
                return false;
            } catch (IOException e) {
                log.warn("发送消息到连接 {} 失败: {}", entry.getKey(), e.getMessage());
                return true; // 移除失败的连接
            }
        });
    }

    /**
     * 向特定连接发送消息
     * @param connectionId 连接ID
     * @param data 数据
     */
    public void sendToConnection(String connectionId, RealtimeData data) {
        SseEmitter emitter = connections.get(connectionId);
        if (emitter == null) {
            log.warn("连接不存在: {}", connectionId);
            return;
        }

        data.setNodeId(nodeId);
        String jsonMessage = JSON.toJSONString(data);

        try {
            emitter.send(SseEmitter.event()
                .id(data.getId())
                .name(data.getType())
                .data(jsonMessage));
        } catch (IOException e) {
            log.error("发送消息到连接 {} 失败: {}", connectionId, e.getMessage());
            connections.remove(connectionId);
        }
    }

    /**
     * 获取活跃连接数
     * @return 连接数
     */
    public int getActiveConnectionCount() {
        return connections.size();
    }

    /**
     * 获取节点ID
     * @return 节点ID
     */
    public String getNodeId() {
        return nodeId;
    }

    /**
     * 生成连接ID
     * @param clientId 客户端ID
     * @return 连接ID
     */
    private String generateConnectionId(String clientId) {
        return String.format("%s-%s-%d", 
            clientId != null ? clientId : "anonymous", 
            nodeId, 
            connectionCounter.incrementAndGet());
    }

    /**
     * 发送欢迎消息
     * @param emitter SSE发射器
     * @param connectionId 连接ID
     */
    private void sendWelcomeMessage(SseEmitter emitter, String connectionId) {
        try {
            RealtimeData welcomeData = RealtimeData.builder()
                .id("welcome-" + System.currentTimeMillis())
                .type(RealtimeData.MessageType.SYSTEM_INFO.getCode())
                .content("连接建立成功，连接ID: " + connectionId)
                .timestamp(LocalDateTime.now())
                .nodeId(nodeId)
                .build();

            emitter.send(SseEmitter.event()
                .id(welcomeData.getId())
                .name("welcome")
                .data(JSON.toJSONString(welcomeData)));
        } catch (IOException e) {
            log.error("发送欢迎消息失败: {}", e.getMessage());
        }
    }

    /**
     * 启动心跳检测
     */
    private void startHeartbeat() {
        heartbeatExecutor.scheduleAtFixedRate(() -> {
            if (!connections.isEmpty()) {
                RealtimeData heartbeat = RealtimeData.builder()
                    .id("heartbeat-" + System.currentTimeMillis())
                    .type(RealtimeData.MessageType.HEARTBEAT.getCode())
                    .content(MapUtils.of(
                        "activeConnections", connections.size(),
                        "nodeId", nodeId,
                        "timestamp", LocalDateTime.now()
                    ))
                    .timestamp(LocalDateTime.now())
                    .nodeId(nodeId)
                    .build();

                broadcast(heartbeat);
            }
        }, 30, 30, TimeUnit.SECONDS);
    }

    /**
     * 清理资源
     */
    @PreDestroy
    public void cleanup() {
        log.info("清理SSE连接管理器资源...");
        
        // 关闭所有连接
        connections.values().forEach(emitter -> {
            try {
                emitter.complete();
            } catch (Exception e) {
                log.warn("关闭SSE连接失败: {}", e.getMessage());
            }
        });
        connections.clear();
        
        // 关闭心跳执行器
        heartbeatExecutor.shutdown();
        try {
            if (!heartbeatExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
                heartbeatExecutor.shutdownNow();
            }
        } catch (InterruptedException e) {
            heartbeatExecutor.shutdownNow();
            Thread.currentThread().interrupt();
        }
        
        log.info("SSE连接管理器资源清理完成");
    }
}